package util

import (
	"go.etcd.io/bbolt"
	"google.golang.org/protobuf/proto"
	"log/slog"
	"statsphere/model"
	"statsphere/pb"
	"statsphere/pkg/cache"
	"statsphere/util/convert"
	"strings"
	"time"
)

func CloseBolt() {
	_ = cache.Bolt.Sync()
	_ = cache.Bolt.Close()
}

var (
	MqReq  = make(chan *model.ReverseCtrlStruct, 4)
	MqResp = make(chan *model.CtrlResultStruct, 2)
)

func processCtrlData() {
	// TODO 超时响应问题 select 里面新增一个case

	// github.com/golang/protobuf 被官方废弃
	// time.DateTime  "2006-01-02 15:04:05"
	var err error

	// 101_1723434893000-20-2131231321

	for {
		select {
		case a := <-MqReq:
			tempId := a.ReqId
			idx := strings.IndexByte(tempId, '-')
			if idx < 1 {
				slog.Error("Id异常,strings.IndexByte")
				continue
			}
			batchId := tempId[0:idx]
			ptIds := a.GetPtIds()
			//
			slog.Debug("反控请求", "batchId", batchId, "pt_ids", ptIds)
			//
		case a := <-MqResp:
			tempId := a.RespId
			idx := strings.IndexByte(tempId, '-')
			if idx < 1 {
				slog.Error("Id异常,strings.IndexByte")
				continue
			}
			batchId := tempId[0:idx]
			ptIds, errItems := a.GetPtIds()
			//
			_bucketName := []byte("log" + time.UnixMilli(a.Time).Format("20060102"))
			_key_id := []byte("id" + batchId)
			ctrlLog := new(pb.CtrlLog)
			//
			isEnd := PopPtIdList(batchId, ptIds, errItems)
			if len(errItems) > 0 {
				err = cache.Bolt.Update(func(tx *bbolt.Tx) error {
					b := tx.Bucket(_bucketName)
					if b == nil {
						b, err = tx.CreateBucket(_bucketName)
						if err != nil {
							return err
						}
					}
					//
					buf := b.Get(_key_id)
					if buf != nil && len(buf) > 0 {
						err = proto.Unmarshal(buf, ctrlLog)
						if err != nil {
							return err
						}
						ctrlLog.Items = append(ctrlLog.Items, errItems...)
					} else {
						ctrlLog.BatchId = batchId
						ctrlLog.Items = errItems
					}
					ctrlLog.IsEnd = isEnd
					buf, _ = proto.Marshal(ctrlLog)
					err = b.Put(_key_id, buf)
					if err != nil {
						return err
					}
					return nil
				})
				if err != nil {
					slog.Error("Bolt_err" + err.Error())
				}
				err = cache.Bolt.Sync()
				if err != nil {
					slog.Error("Bolt_err_Sync" + err.Error())
				}

			}

			if isEnd == true {
				// 当前没有错误的时候
				err = cache.Bolt.Update(func(tx *bbolt.Tx) error {
					b := tx.Bucket(_bucketName)
					if b == nil {
						b, err = tx.CreateBucket(_bucketName)
						if err != nil {
							return err
						}
					}
					//
					buf := b.Get(_key_id)
					if buf != nil && len(buf) > 0 {
						err = proto.Unmarshal(buf, ctrlLog)
						if err != nil {
							return err
						}
					} else {
						ctrlLog.BatchId = batchId
						ctrlLog.Items = nil
					}
					ctrlLog.IsEnd = true
					buf, _ = proto.Marshal(ctrlLog)
					err = b.Put(_key_id, buf)
					if err != nil {
						return err
					}
					return nil
				})
				if err != nil {
					slog.Error("Bolt_err" + err.Error())
				}
				err = cache.Bolt.Sync()
				if err != nil {
					slog.Error("Bolt_err_Sync" + err.Error())
				}
				// 反控成功 失败详情
				go func() {
					cache.Rdb0.Incr(ctx, "ctrl_count:all")
					if ctrlLog.Items != nil && len(ctrlLog.Items) > 0 {
						slog.Error("反控失败,详情", "batch_id", ctrlLog.BatchId, "items", ctrlLog.Items)
						_ = cache.Rdb0.Incr(ctx, "ctrl_count:err")
					} else {
						slog.Info("反控成功,batch_id:" + ctrlLog.BatchId)
						_ = cache.Rdb0.Incr(ctx, "ctrl_count:suc")
					}
				}()

				// mqtt上传规则日志
				ruleCtrlLog := convert.ConvertToRuleCtrlLog(ctrlLog)
				pushLog(ruleCtrlLog)

			} else {
				// 啥也不操作 isEnd = false
			}
		}
	}
	//
}
